package com.hdaccp.log
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._

import scala.collection.mutable.ListBuffer
object Demo2 {
  def main(args: Array[String]): Unit = {
    //得到sparkSession对象
    val spark = SparkSession.builder().appName("TopNStatJob")
      .config("spark.sql.sources.partitionColumnTypeInference.enabled","false")
      .master("local[2]").getOrCreate()

  //从文件转换成DataFrame对象
    val accessDF = spark.read.format("parquet").load("F:/accp教学/sparkresources/clean5")
    //    accessDF.printSchema()
    //    accessDF.show(false)
    val day = "20180815"
    //按照地市进行统计TopN课程
    cityAccessTopNStat(spark, accessDF, day)

    spark.stop()
  }

  /**
    * 按照地市进行统计TopN课程
    */
def cityAccessTopNStat(spark: SparkSession, accessDF:DataFrame, day:String): Unit = {
    import spark.implicits._

   /*  val cityAccessTopNDF = accessDF.filter($"day" === day && $"cmsType" === "class")
      .groupBy("day","cmsId")
      .agg(count("cmsId").as("times")).orderBy(count("cmsId").desc).limit(3)*/


   val cityAccessTopNDF = accessDF.filter($"day" === day && $"cmsType" === "class")
    .groupBy("day","cmsId")
    .agg(count("cmsId").as("times"))
   // cityAccessTopNDF.show(false)

    //Window函数在Spark SQL的使用

    val top3DF = cityAccessTopNDF.select(
      cityAccessTopNDF("day"),
    //  cityAccessTopNDF("city"),
      cityAccessTopNDF("cmsId"),
      cityAccessTopNDF("times"),
      row_number().over(Window.partitionBy(cityAccessTopNDF("day"))
        .orderBy(cityAccessTopNDF("times").desc)
      ).as("times_rank")
    ).filter("times_rank <=3") //.show(false)  //Top3

 //   top3DF.show(false)
    /**
      * 将统计结果写入到MySQL中
      */


   try {
     top3DF.foreachPartition(partitionOfRecords => {
        val list = new ListBuffer[DayCityVideoAccessStat]

        partitionOfRecords.foreach(info => {
          val day = info.getAs[String]("day")
          val cmsId = info.getAs[String]("cmsId")
          val times = info.getAs[Long]("times")
          val timesRank = info.getAs[Int]("times_rank")
          list.append(DayCityVideoAccessStat(day, cmsId, times, timesRank))
        })

        StatDAO.insertDayCityVideoAccessTopN(list)
      })
    } catch {
      case e:Exception => e.printStackTrace()
    }

  }

}
